package cn.edu.cug.cs.gtl.gsm.aof.hdfs;

import cn.edu.cug.cs.gtl.common.Identifier;
import cn.edu.cug.cs.gtl.gsm.aof.DataFiles;
import cn.edu.cug.cs.gtl.gsm.item.DataItem;
import cn.edu.cug.cs.gtl.gsm.item.IndexMap;
import cn.edu.cug.cs.gtl.gsm.item.MetaItem;
import org.apache.hadoop.fs.*;
import java.io.DataInputStream;
import java.io.DataOutputStream;
import java.io.IOException;

public class HDFSDataFiles implements DataFiles {
    private String fileName = "hdfs_sm_data";
    private FileSystem fileSystem;
    private String workDirectory;
    private MetaItem metaItem;
    private IndexMap indexMap;

    public HDFSDataFiles(FileSystem fileSystem, String workDirectory, MetaItem metaItem, IndexMap indexMap) {
        this.fileSystem = fileSystem;
        this.workDirectory = workDirectory;
        this.metaItem = metaItem;
        this.indexMap = indexMap;
        fileName = "hdfs_sm_data_" + this.metaItem.getBaseName() + "_0";
    }

    @Override
    public void setCurrentFileOrder(int order) {
        fileName = "hdfs_sm_data_" + this.metaItem.getBaseName() + "_" + String.valueOf(order);
    }

    @Override
    public boolean create() throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        fileSystem.createNewFile(new Path(fullPathName));
        return true;
    }

    @Override
    public boolean drop() throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        boolean b = fileSystem.deleteOnExit(new Path(fullPathName));
        if (b) {
            System.out.println("删除成功");
        } else {
            System.out.println("删除失败");
        }
        return b;
    }

    @Override
    public boolean exists() throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        return HDFSStorageManager.exitsFile(fileSystem, fullPathName);
    }

    @Override
    public boolean write(DataItem dataItem) throws IOException {
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        FSDataOutputStream fsDataOutputStream = null;
        if (fileSystem.getContentSummary(new Path(fullPathName)).getLength() > 0) {
             fsDataOutputStream = fileSystem.append(new Path(fullPathName));
        } else if(fileSystem.getContentSummary(new Path(fullPathName)).getLength() == 0){
             fsDataOutputStream = fileSystem.create(new Path(fullPathName));
        }
        fsDataOutputStream.write(dataItem.getPageContent());
        fsDataOutputStream.close();
        return true;
    }

    @Override
    public DataItem read(Identifier pageIdentifier) throws IOException {
        int order = metaItem.getCollectionOrder(pageIdentifier);
        setCurrentFileOrder(order);
        String fullPathName = HDFSStorageManager.concatenate(workDirectory, metaItem.getBaseName(), fileName);
        long pos = pageIdentifier.longValue()-metaItem.getCollectionSize()* order;
        pos = pos * metaItem.getPageSize();
        FSDataInputStream open = fileSystem.open(new Path(fullPathName));
        DataInputStream dataInputStream = new DataInputStream(open);
        byte[] bytes = new byte[metaItem.getPageSize()];
        dataInputStream.skipBytes((int) pos);
        dataInputStream.readFully(bytes);
        return new DataItem(pageIdentifier, bytes);
    }
}
